package com.atguigu.Func;


import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.Util.DimUtil_11;
import com.atguigu.Util.DruidDSUtil;
import com.atguigu.Util.ThreadPoolUtil11;
import lombok.SneakyThrows;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;

/**
 * @author hjy
 * @create 2023/3/21 11:03
 */
public abstract class AsyncDimFunction_11<T> extends RichAsyncFunction<T,T> implements AsyncJoinFunction_11<T> {
    //异步io
    private String tableName;

    public AsyncDimFunction_11(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {
        ThreadPoolUtil11.getThreadPoolExecutor().execute(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
                String id = getId(input);
                JSONObject dimInfo = DimUtil_11.getDimInfo(phoenixConn, tableName, id);

                joinDimInfo(input,dimInfo);
                resultFuture.complete(Collections.singleton(input));

                phoenixConn.close();
            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("超时数据:"+input);
    }
}
